package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumClientFeign;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserClientFeign;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.search.AlbumSearchVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@Slf4j
@Service
public class SearchServiceImpl implements SearchService {

    @Autowired
    AlbumClientFeign albumClientFeign;
    @Autowired
    UserClientFeign userClientFeign;
    @Autowired
    ElasticsearchClient elasticsearchClient;
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 搜索首页：三级分类TOP7中的专辑TOP6
     *
     * @param category1Id
     * @return
     */
    @Override
    @SneakyThrows
    public Object channel(Long category1Id) {
        // 根据一级分类查询其下top7的三级分类
        List<BaseCategory3> baseCategory3List =
                albumClientFeign.findTopBaseCategory3(category1Id);
        // list转换为map：方便返回结果JSON获取参数
        Map<Long, BaseCategory3> baseCategory3Map = baseCategory3List.stream().collect(Collectors.toMap(
                key -> key.getId(),
                value -> value
        ));
        // 遍历映射查询条件-三级分类idList - FieldValue类型
        List<FieldValue> category3IdList = baseCategory3List.stream().map(baseCategory3 -> {
            return FieldValue.of(baseCategory3.getId());
        }).collect(Collectors.toList());

        // 查询条件初始化
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 指定查询的索引
        builder.index("albuminfo_0508");
        // 拼接数据的筛选条件：获取当前一级分类id下三级分类TOP7
        builder.query( // 执行查询
                query -> query.terms(  // 进行词条查询
                        terms -> terms.field("category3Id").terms( // 指定查询的域
                                queryCondition -> queryCondition.value(category3IdList) // 指定查询条件：分类id list
                        )
                )
        );

        // 构建分桶语句：1.主聚合查询:根据3级id分桶 2.子聚合查询:2.1根据热度排序 2.2 降序 取前6
        // 聚合查询(分桶)：
        builder.aggregations(
                "aggCategory3Id", // 定义别名
                agg -> agg.terms(
                        terms -> terms.field("category3Id") // 聚合(分桶)条件：指定域
                ).aggregations( // 子聚合查询
                        "topHits",
                        subAgg -> subAgg.topHits( // 优先命中
                                topHits -> topHits
                                        .sort(
                                                sort -> sort.field(
                                                        field -> field.field("hotScore").order(SortOrder.Desc) // 指定排序条件域，排序方式
                                                )
                                        )
                                        .size(6) // 指定size
                        )
                )
        );

        // 执行查询：去es中查询 - elasticsearchClient
        SearchResponse<AlbumInfoIndex> searchResponse =
                elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);

        // 获取结果：查询结果(hits) + 聚合结果(aggregations)
        // 查询聚合结果
        Map<String, Aggregate> aggregations = searchResponse.aggregations();
        if (aggregations != null) { //校验非空
            // 获取全部三级分类桶的聚合结果
            Aggregate aggregate = aggregations.get("aggCategory3Id");

            // 返回：解析主聚合&子聚合后方法返回
            return aggregate.lterms().buckets().array().stream().map(longTermsBucket -> {
                // 获取对象值--分桶条件
                long category3Id = longTermsBucket.key();
                // 包装返回结果
                JSONObject jsonObject = new JSONObject();
                // 三级分类id
                jsonObject.put("baseCategory3", baseCategory3Map.get(category3Id));

                // 获取子聚合
                Map<String, Aggregate> subAggregate = longTermsBucket.aggregations();
                if (subAggregate != null) {
                    // 获取别名结果
                    Aggregate topHitsInfoAgg = subAggregate.get("topHits");
                    // 获取桶中数据：数据处理 agg --> hit --> map --> list --> 遍历映射
                    List<AlbumInfoIndex> albumInfoIndexList =
                            topHitsInfoAgg.topHits().hits().hits().stream().map(album -> {
                                return album.source().to(AlbumInfoIndex.class);
                            }).collect(Collectors.toList());
                    jsonObject.put("list", albumInfoIndexList);
                }
                // 返回解析的聚合查询结果
                return jsonObject;
            }).collect(Collectors.toList());

        }
        // 如果聚合查询结果为空则返null
        return null;
    }


    /**
     * 根据条件搜索专辑
     *
     * @param albumIndexQuery
     * @return
     */
    @Override
    @SneakyThrows
    public Object search(AlbumIndexQuery albumIndexQuery) {
        // 查询构造
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 指定索引
        builder.index("albuminfo_0508");
        // 查询条件:私有方法
        builder = buildSearchParams(builder, albumIndexQuery);
        // 执行查询
        SearchResponse<AlbumInfoIndex> searchResponse =
                elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);
        // 解析结果:私有方法
        AlbumSearchResponseVo searchResponseVo = getSearchResult(searchResponse);
        // 补全VO属性
        searchResponseVo.setPageNo(albumIndexQuery.getPageNo());
        Integer pageSize = albumIndexQuery.getPageSize();
        searchResponseVo.setPageSize(pageSize);
        Long total = searchResponseVo.getTotal();
        searchResponseVo.setTotalPages(total % pageSize == 0 ? (total / pageSize) : ((total / pageSize) + 1));
        // 返回结果
        return searchResponseVo;
    }


    /**
     * 提示词搜索
     * @param keywords
     * @return
     */
    @SneakyThrows
    @Override
    public Object completeSuggest(String keywords) {
        // 拼接的查询条件：前缀匹配
        // 创建查询builder
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 设置索引
        builder.index("suggestinfo_0508");
        // 构建提示词条件 -- suggest接口
        builder.suggest(suggest ->
                suggest
                        .suggesters(
                                "suggestKeyword",
                                sug -> sug.prefix(keywords).completion(
                                        c -> c.field("keyword") // 匹配的域
                                                .size(10) // 返回一个list 最多10个元素(可视为桶)
                                                .skipDuplicates(true) // 去重
                                                .fuzzy(f -> f.fuzziness("auto"))  // 固定写法 提示的内容
                                )
                        )
                        .suggesters(
                                "suggestPinyin",
                                sug -> sug.prefix(keywords).completion(
                                        c -> c.field("keywordPinyin")
                                                .size(10)
                                                .skipDuplicates(true) // 去重
                                                .fuzzy(f -> f.fuzziness("auto"))
                                )
                        )
                        .suggesters(
                                "suggestSequence",
                                sug -> sug.prefix(keywords).completion(
                                        c -> c.field("keywordSequence")
                                                .size(10)
                                                .skipDuplicates(true) // 去重
                                                .fuzzy(f -> f.fuzziness("auto"))
                                )
                        )

        );
        // 执行搜索 -- 查询的类型
        SearchResponse<SuggestIndex> searchResponse =
                elasticsearchClient.search(builder.build(), SuggestIndex.class);
        // 获取查询结果 -- suggest查询
        Map<String, List<Suggestion<SuggestIndex>>> suggestMap = searchResponse.suggest();
        // 解析查询结果
        if (suggestMap != null) {
            // 解析汉字
            List<String> list1 = getSuggestResult(suggestMap, "suggestKeyword");
            // 解析配音
            List<String> list2 = getSuggestResult(suggestMap, "suggestPinyin");
            // 解析首字母
            List<String> list3 = getSuggestResult(suggestMap, "suggestSequence");
            // 处理数据 09:33
            // 返回三个list的合并结果
            return Stream.of(list1, list2, list3).flatMap(Collection::stream)
                    .collect(Collectors.toSet()) //set去重
                    .stream() //转换为流
                    .limit(10) // 截取10条
                    .collect(Collectors.toList());
        }
        // 返回结果
        return null;
    }


    /**
     * 查询专辑的详细信息
     *
     * @param albumId
     * @return
     */
    @Override
    @SneakyThrows
    public Object getAlbumInfo(Long albumId) {
        // 返回结果
        Map<String, Object> result = new ConcurrentHashMap<>();
        // 布隆过滤器FeignClient查询 有则查询 无则直接return result
        if (albumClientFeign.getAlbumFromBloom(albumId)) {
            // 根据id查询专辑
            CompletableFuture<AlbumInfo> future1 = CompletableFuture.supplyAsync(() -> {
                AlbumInfo albumInfo = albumClientFeign.getAlbumInfo(albumId);
                if (albumInfo == null || albumInfo.getId() == null) {
                    return null;
                }
                result.put("albumInfo", albumInfo);
                return albumInfo;
            }, threadPoolExecutor);
            // 分类
            CompletableFuture<Void> future2 = future1.thenAcceptAsync(albumInfo -> {
                if (albumInfo == null || albumInfo.getId() == null) {
                    return;
                }
                BaseCategoryView baseCategoryView =
                        albumClientFeign.getBaseCategoryView(albumInfo.getCategory3Id());
                result.put("baseCategoryView", baseCategoryView);
            }, threadPoolExecutor);
            // 统计
            CompletableFuture<Void> future3 = future1.thenAcceptAsync(albumInfo -> {
                if (albumInfo == null || albumInfo.getId() == null) {
                    return;
                }
                Map<String, Object> albumStatMap = albumClientFeign.getAlbumStatMap(albumId);
                AlbumStatVo albumStatVo = new AlbumStatVo();
                albumStatVo.setAlbumId(albumId);
                albumStatVo.setPlayStatNum(Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_PLAY).toString()));
                albumStatVo.setSubscribeStatNum(Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_SUBSCRIBE).toString()));
                albumStatVo.setPlayStatNum(Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_BROWSE).toString()));
                albumStatVo.setCommentStatNum(Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_COMMENT).toString()));
                result.put("albumStatVo", albumStatVo);
            }, threadPoolExecutor);
            // 作者
            CompletableFuture<Void> future4 = future1.thenAcceptAsync(albumInfo -> {
                if (albumInfo == null || albumInfo.getId() == null) {
                    return;
                }
                UserInfo userInfo = userClientFeign.getUserInfo(albumInfo.getUserId());
                UserInfoVo userInfoVo = new UserInfoVo();
                BeanUtils.copyProperties(userInfo, userInfoVo);
                result.put("announcer", userInfoVo);
            }, threadPoolExecutor);
            // 等待所有副线程结束    4个异步编排 Requests per second:约500
            CompletableFuture.allOf(future2, future3, future4).join();
//        future1.get();   1个异步编排 Requests per second:约3200
            // 直接请求 Requests per second:约7000
            // 测试用 1个feign调用 Requests per second:约3200 递减
//        AlbumInfo info = albumClientFeign.getAlbumInfo(albumId);
//        return info ;
        }
        return result;
    }

    /**
     * 解析查询关键词
     * @param suggestMap
     * @param name
     * @return
     */
    private CopyOnWriteArrayList<String> getSuggestResult(
            Map<String, List<Suggestion<SuggestIndex>>> suggestMap,
            String name) {
        // 初始化
        CopyOnWriteArrayList<String> listResult = new CopyOnWriteArrayList<>();
        // 解析匹配结果并记录
        suggestMap.get(name).get(0).completion().options().stream().forEach(c -> {
            listResult.add(c.source().getTitle());
        });
        // 返回解析结果
        return listResult;
    }


    /**
     * 专辑搜索条件拼接
     *
     * @param builder
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest.Builder buildSearchParams(SearchRequest.Builder builder,
                                                    AlbumIndexQuery albumIndexQuery) {
        // 构建组合查询
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        // query-filter
        // 获取关键字条件-query对象获取
        String keyword = albumIndexQuery.getKeyword();
        //非空
        if (StringUtils.isNotEmpty(keyword)) {
            boolQueryBuilder // 且查询
                    .should(should -> should.match(match -> match.field("albumTitle").query(keyword))) // 专辑标题
                    .should(should -> should.match(match -> match.field("albumIntro").query(keyword))) // 专辑描述
                    .should(should -> should.match(match -> match.field("announcerName").query(keyword))); // 专辑作者
        }
        // 过滤：1级分类
        Long category1Id = albumIndexQuery.getCategory1Id();
        if (category1Id != null) {
            boolQueryBuilder.filter(filter -> filter.term(
                    term -> term.field("category1Id").value(category1Id)
            ));
        }
        // 过滤：2级分类
        Long category2Id = albumIndexQuery.getCategory2Id();
        if (category2Id != null) {
            boolQueryBuilder.filter(filter -> filter.term(
                    term -> term.field("category2Id").value(category2Id)
            ));
        }
        // 过滤：3级分类
        Long category3Id = albumIndexQuery.getCategory3Id();
        if (category3Id != null) {
            boolQueryBuilder.filter(filter -> filter.term(
                    term -> term.field("category3Id").value(category3Id)
            ));
        }
        // 标签
        List<String> attributeList = albumIndexQuery.getAttributeList();
        // 非空
        if (attributeList != null && attributeList.size() > 0) {
            // 遍历条件拼接
            attributeList.stream().forEach(attribute -> {
                // 属性id:属性值
                String[] split = attribute.split(":");
                // 添加过滤
                boolQueryBuilder.filter(
                        filter -> filter.nested(
                                nested -> nested
                                        .path("attributeValueIndexList")
                                        .query(query -> query.bool( // 设置查询
                                                bool -> bool // 设置bool查询
                                                        .must(must -> must    // 且查询
                                                                .term(term -> term
                                                                        .field("attributeValueIndexList.attributeId")
                                                                        .value(split[0])))
                                                        .must(must -> must    // 且查询
                                                                .term(term -> term
                                                                        .field("attributeValueIndexList.valueId")
                                                                        .value(split[1])))
                                        ))
                        )
                );
            });

        }
        // 保存查询条件
        builder.query(boolQueryBuilder.build()._toQuery());
        // 分页
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        builder.from((pageNo - 1) * pageSize); // 指定开始条数
        builder.size(pageSize);
        // 高亮
        builder.highlight(highlight -> highlight.fields(
                "albumTitle", // 高亮部分
                hl -> hl.preTags("<font style=color:red>").postTags("</font>") // 拼标签
        ));
        // 分页排序条件: 1 综合 id  2 播放 playStatNum 3 发布时间 createTime
        String order = albumIndexQuery.getOrder();
        if (StringUtils.isNotEmpty(order)) {
            String[] split = order.split(":");
            switch (split[0]) {
                case "1" -> builder.sort(
                        sort -> sort.field(field -> field.field("id").order(SortOrder.Desc)));
                case "2" -> builder.sort(
                        sort -> sort.field(field -> field.field("playStatNum").order(SortOrder.Desc)));
                case "3" -> builder.sort(
                        sort -> sort.field(field -> field.field("createTime").order(SortOrder.Desc)));
            }
        } else {
            // 默认
            builder.sort(
                    sort -> sort.field(field -> field.field("hotScore").order(SortOrder.Desc)));
        }
        return builder;
    }


    /**
     * 专辑搜索结果解析
     *
     * @param searchResponse
     * @return
     */
    private AlbumSearchResponseVo getSearchResult(SearchResponse<AlbumInfoIndex> searchResponse) {
        // 初始化返回结果
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        // 获取符合查询数据
        HitsMetadata<AlbumInfoIndex> hits = searchResponse.hits();
        // 获取数据数量
        long total = hits.total().value();
        albumSearchResponseVo.setTotal(total);
        // 解析
        List<AlbumInfoIndexVo> albumInfoIndexVoList = hits.hits().stream().map(hit -> {
            // 获取原始数据
            AlbumInfoIndex albumInfoIndex = hit.source();
            // 获取高亮
            Map<String, List<String>> highlight = hit.highlight();
            // 非空
            if (highlight != null) {
                // 设置高亮
                List<String> highlightList = highlight.get("albumTitle");
                // 非空
                if (highlightList != null) {
                    String albumTitle =
                            highlightList.stream().collect(Collectors.joining()); // 拼接高亮
                    albumInfoIndex.setAlbumTitle(albumTitle);
                }
            }
            // 返回结果初始化
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            // 复制属性
            return albumInfoIndexVo;
        }).collect(Collectors.toList());
        // 返回
        albumSearchResponseVo.setList(albumInfoIndexVoList);
        return albumSearchResponseVo;
    }


    /**
     * 查询指定一级分类下指定项目前10专辑返回前端
     * @param category1Id
     * @return
     */
    @Override
    public List<AlbumInfoIndexVo> findRankingList(String category1Id,String item) {
        List<AlbumInfoIndexVo> albumInfoIndexVoList = (List<AlbumInfoIndexVo>)redisTemplate.opsForHash().get(category1Id, item);
        if (albumInfoIndexVoList != null){
            return albumInfoIndexVoList;
        }
        List<AlbumInfoIndex> albumInfoIndexListFromES = rank(category1Id, item);
        List<AlbumInfoIndexVo> albumInfoIndexVoListFromES = albumInfoIndexListFromES.stream().map(albumInfoIndex -> {
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            return albumInfoIndexVo;
        }).collect(Collectors.toList());
        redisTemplate.opsForHash().put(category1Id+"",item,albumInfoIndexVoListFromES);
        return null;
    }

    /**
     * 从ES中查询指定category1Id下对应的item排行榜
     * @param category1Id
     * @param item
     */
    @SneakyThrows
    @Override
    public List<AlbumInfoIndex> rank(String category1Id, String item) {
        // 初始化查询
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 指定查询的索引
        builder.index("albuminfo_0508");
        // 构建分词查询
        builder.query(q -> q.term(t -> t.field("category1Id").value(category1Id + "")));
        // 构建聚合查询
        builder.aggregations(
                // 别名
                "aggCategory1Id",
                // 返回顶部文档 -- 排序 -- 以item字段为条件 -- 按降序 -- 限10条
                a -> a.topHits(t -> t.sort(s -> s.field(f -> f.field(item).order(SortOrder.Desc))).size(10))
        );

        // 执行查询：去es中查询 - elasticsearchClient
        SearchResponse<AlbumInfoIndex> searchResponse =
                elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);
        // 查询聚合结果
        Map<String, Aggregate> aggregations = searchResponse.aggregations();
        if (aggregations != null) { //校验非空
            // 获取全部1级分类桶的聚合结果
            Aggregate aggregate = aggregations.get("aggCategory1Id");
            // 返回：解析主聚合后方法返回
            return aggregate.topHits().hits().hits().stream().map(albumInfoIndex -> {
                return albumInfoIndex.source().to(AlbumInfoIndex.class);
            }).collect(Collectors.toList());
        }
        return null;
    }
}
